package com.chukun.flink.table.type;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;

/**
 * @author chukun
 * @version 1.0.0
 * @description 使用类型
 * @createTime 2022年05月29日 22:11:00
 */
public class RowRegisterRowWithNames {

    public static void main(String[] args) throws Exception {

        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建table的环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        List<Row> data = new ArrayList<>();
        data.add(Row.of(1, 1L, "Hi"));
        data.add(Row.of(2, 2L, "Hello"));
        data.add(Row.of(3, 2L, "Hello world"));

        // 定义数据类型
        TypeInformation<?>[] types = new TypeInformation[] {
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.LONG_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO
        };

        // 定义数据字段
        String [] names = new String[] {"a", "b", "c"};

        // 创建RowType对象
        RowTypeInfo rowTypeInfo = new RowTypeInfo(types, names);

        // 创建dataStream
        SingleOutputStreamOperator<Row> originalStream = env.fromCollection(data).returns(rowTypeInfo);

        // 注册临时表，表名为 tableRow, 数据信息为 originalStream
        tableEnv.createTemporaryView("tableRow", originalStream);

        // 书写sql语句
        String sql = "select a,b from tableRow";

        // 执行sql，返回一张结果表
        Table resultTable = tableEnv.sqlQuery(sql);

        // 结果表转为数据流
        DataStream<Row> resultDataStream = tableEnv.toDataStream(resultTable, Row.class);

        resultDataStream.print("toDataStream");

        env.execute();
    }
}
